package com.atguigu.flink.chapter08_exec2;

import com.atguigu.flink.pojo.MyUtil;
import com.atguigu.flink.pojo.UserBehavior;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * Created by Smexy on 2022/10/29
 *
 * 每隔5分钟输出最近1小时内点击量最多的前N个商品
 *
 *  计算方式: TopN
 *      全量运算： 获取到计算的所有数据，再排序后，求topN
 *
 *      窗口: slide = 5min
 *            size = 1h
 *            滑动窗口
 *
 *      ①统计，每个窗口(规定时间范围)内所有商品，各自的点击量
 *          10：00 - 11：00 ：
 *                                  a: 100, b:101, c:102 ....
 *          10:05  - 11: 05 :
 *                                   a: 100, d:101, c:102 ....
 *
 *            聚合运算:  汇总某个窗口中某个商品的总点击量
 *                          按照商品id分组，求和。
 *                              增量聚合:
 *                                          reduce,aggregate(选)
 *
 *       ②针对同一个时间窗口内，不同商品的点击量，排序求TopN
 *                           按照窗口id分组，求和
 *
 *            如何保证上游的窗口计算已经全部完成，数据已经推荐到下游？
 *                  只要下游的watermark到达2点，说明上游的时间已经全部在2点及以后。
 *                  因为下游取上游并行度中watermark最小的，此时就可以出发运算。
 *
 *                  下游在watermark到2点，准备触发topN的运算。
 *
 窗口[2017-11-26 08:15:00,2017-11-26 09:15:00):[812879:7, 3624285:5, 2338453:5]
 窗口[2017-11-26 08:05:00,2017-11-26 09:05:00):[4261030:3, 5051027:3, 3493253:3]
 窗口[2017-11-26 08:50:00,2017-11-26 09:50:00):[2338453:24, 812879:14, 2563440:14]
 窗口[2017-11-26 08:55:00,2017-11-26 09:55:00):[2338453:25, 2563440:18, 812879:16]
 窗口[2017-11-26 08:10:00,2017-11-26 09:10:00):[812879:5, 3624285:4, 4261030:4]
 窗口[2017-11-26 08:45:00,2017-11-26 09:45:00):[2338453:20, 812879:14, 2563440:14]
 窗口[2017-11-26 08:20:00,2017-11-26 09:20:00):[812879:8, 2338453:8, 2563440:7]
 窗口[2017-11-26 08:40:00,2017-11-26 09:40:00):[2338453:20, 812879:14, 2563440:12]
 窗口[2017-11-26 08:25:00,2017-11-26 09:25:00):[2338453:13, 812879:11, 2563440:7]
 窗口[2017-11-26 08:35:00,2017-11-26 09:35:00):[2338453:17, 812879:11, 2331370:9]
 窗口[2017-11-26 08:30:00,2017-11-26 09:30:00):[2338453:15, 812879:11, 2563440:8]
 窗口[2017-11-26 09:00:00,2017-11-26 10:00:00):[2338453:27, 2563440:18, 812879:17]
 窗口[2017-11-26 09:10:00,2017-11-26 10:10:00):[2338453:30, 812879:18, 2563440:14]
 窗口[2017-11-26 09:25:00,2017-11-26 10:25:00):[2338453:32, 812879:16, 3244134:16]
 窗口[2017-11-26 09:55:00,2017-11-26 10:55:00):[2338453:36, 812879:19, 3845720:16]
 窗口[2017-11-26 09:05:00,2017-11-26 10:05:00):[2338453:31, 812879:18, 2563440:16]
 窗口[2017-11-26 09:45:00,2017-11-26 10:45:00):[2338453:34, 812879:17, 3845720:17]
 窗口[2017-11-26 10:05:00,2017-11-26 11:05:00):[2338453:31, 812879:18, 3845720:17]
 窗口[2017-11-26 10:25:00,2017-11-26 11:25:00):[2338453:28, 812879:16, 3034696:15]
 窗口[2017-11-26 09:15:00,2017-11-26 10:15:00):[2338453:33, 812879:18, 2563440:13]
 窗口[2017-11-26 10:20:00,2017-11-26 11:20:00):[2338453:29, 3034696:18, 812879:15]
 窗口[2017-11-26 09:40:00,2017-11-26 10:40:00):[2338453:33, 812879:17, 3034696:14]
 窗口[2017-11-26 10:15:00,2017-11-26 11:15:00):[2338453:30, 3034696:18, 812879:16]
 窗口[2017-11-26 09:30:00,2017-11-26 10:30:00):[2338453:32, 812879:18, 259923:16]
 窗口[2017-11-26 09:50:00,2017-11-26 10:50:00):[2338453:36, 812879:18, 3845720:18]
 窗口[2017-11-26 10:00:00,2017-11-26 11:00:00):[2338453:35, 812879:20, 3845720:17]
 窗口[2017-11-26 09:20:00,2017-11-26 10:20:00):[2338453:32, 812879:18, 3244134:15]
 窗口[2017-11-26 09:35:00,2017-11-26 10:35:00):[2338453:35, 812879:19, 3034696:16]
 窗口[2017-11-26 10:10:00,2017-11-26 11:10:00):[2338453:33, 3845720:17, 812879:16]
 窗口[2017-11-26 11:45:00,2017-11-26 12:45:00):[2338453:24, 3810981:21, 4443059:19]
 窗口[2017-11-26 11:00:00,2017-11-26 12:00:00):[2338453:28, 987143:17, 1583704:17]
 窗口[2017-11-26 11:30:00,2017-11-26 12:30:00):[2338453:25, 4443059:16, 2364679:15]
 窗口[2017-11-26 10:55:00,2017-11-26 11:55:00):[2338453:26, 812879:18, 987143:16]
 窗口[2017-11-26 12:05:00,2017-11-26 13:05:00):[2338453:23, 812879:21, 998153:18]
 窗口[2017-11-26 12:00:00,2017-11-26 13:00:00):[2338453:24, 3810981:19, 998153:18]
 窗口[2017-11-26 11:50:00,2017-11-26 12:50:00):[2338453:24, 3810981:22, 812879:16]
 窗口[2017-11-26 11:55:00,2017-11-26 12:55:00):[2338453:25, 3810981:21, 812879:15]
 窗口[2017-11-26 10:30:00,2017-11-26 11:30:00):[2338453:28, 812879:17, 2364679:15]
 窗口[2017-11-26 11:20:00,2017-11-26 12:20:00):[2338453:27, 812879:18, 4443059:18]
 窗口[2017-11-26 11:10:00,2017-11-26 12:10:00):[2338453:26, 4443059:16, 1583704:16]
 窗口[2017-11-26 10:35:00,2017-11-26 11:35:00):[2338453:27, 812879:17, 2364679:17]
 窗口[2017-11-26 10:50:00,2017-11-26 11:50:00):[2338453:26, 812879:19, 1583704:17]
 窗口[2017-11-26 10:45:00,2017-11-26 11:45:00):[2338453:30, 812879:19, 2364679:16]
 窗口[2017-11-26 11:35:00,2017-11-26 12:35:00):[2338453:25, 812879:16, 3810981:16]
 窗口[2017-11-26 11:15:00,2017-11-26 12:15:00):[2338453:25, 4443059:18, 812879:17]
 窗口[2017-11-26 11:05:00,2017-11-26 12:05:00):[2338453:28, 987143:17, 44126:16]
 窗口[2017-11-26 12:10:00,2017-11-26 13:10:00):[2338453:23, 812879:20, 998153:18]
 窗口[2017-11-26 10:40:00,2017-11-26 11:40:00):[2338453:29, 812879:18, 2364679:17]
 窗口[2017-11-26 12:25:00,2017-11-26 13:25:00):[2338453:22, 812879:19, 998153:18]
 窗口[2017-11-26 12:15:00,2017-11-26 13:15:00):[2338453:23, 998153:20, 2364679:17]
 窗口[2017-11-26 12:30:00,2017-11-26 13:30:00):[998153:22, 812879:22, 2338453:21]
 窗口[2017-11-26 11:25:00,2017-11-26 12:25:00):[2338453:24, 4443059:17, 812879:16]
 窗口[2017-11-26 11:40:00,2017-11-26 12:40:00):[2338453:23, 3810981:21, 4443059:17]
 窗口[2017-11-26 12:20:00,2017-11-26 13:20:00):[2338453:23, 998153:19, 812879:18]
 窗口[2017-11-26 12:35:00,2017-11-26 13:35:00):[998153:22, 812879:18, 2338453:17]
 窗口[2017-11-26 13:10:00,2017-11-26 14:10:00):[812879:20, 624903:16, 987143:15]
 窗口[2017-11-26 13:15:00,2017-11-26 14:15:00):[812879:23, 4649427:14, 3520504:14]
 窗口[2017-11-26 13:30:00,2017-11-26 14:30:00):[812879:20, 2338453:18, 3034696:15]
 窗口[2017-11-26 13:00:00,2017-11-26 14:00:00):[812879:16, 624903:15, 3031354:15]
 窗口[2017-11-26 13:05:00,2017-11-26 14:05:00):[624903:16, 987143:16, 812879:15]
 窗口[2017-11-26 13:35:00,2017-11-26 14:35:00):[812879:20, 2338453:18, 3034696:16]
 窗口[2017-11-26 14:05:00,2017-11-26 15:05:00):[812879:25, 1871901:20, 3845720:17]
 窗口[2017-11-26 14:10:00,2017-11-26 15:10:00):[1871901:21, 3845720:20, 812879:20]
 窗口[2017-11-26 12:40:00,2017-11-26 13:40:00):[998153:21, 812879:18, 2338453:17]
 窗口[2017-11-26 13:40:00,2017-11-26 14:40:00):[812879:20, 2338453:18, 3034696:16]
 窗口[2017-11-26 13:20:00,2017-11-26 14:20:00):[812879:21, 4649427:14, 3520504:14]
 窗口[2017-11-26 13:50:00,2017-11-26 14:50:00):[812879:23, 2338453:19, 3845720:18]
 窗口[2017-11-26 14:00:00,2017-11-26 15:00:00):[812879:26, 1871901:19, 3845720:16]
 窗口[2017-11-26 13:55:00,2017-11-26 14:55:00):[812879:25, 3845720:17, 1871901:17]
 窗口[2017-11-26 12:45:00,2017-11-26 13:45:00):[998153:19, 812879:17, 987143:17]
 窗口[2017-11-26 12:55:00,2017-11-26 13:55:00):[812879:18, 624903:14, 4649427:14]
 窗口[2017-11-26 13:45:00,2017-11-26 14:45:00):[812879:21, 2338453:20, 3845720:16]
 窗口[2017-11-26 12:50:00,2017-11-26 13:50:00):[812879:18, 998153:16, 987143:16]
 窗口[2017-11-26 13:25:00,2017-11-26 14:25:00):[812879:20, 2338453:17, 4649427:14]
 窗口[2017-11-26 16:15:00,2017-11-26 17:15:00):[2338453:23, 1850198:21, 2364679:18]
 窗口[2017-11-26 17:20:00,2017-11-26 18:20:00):[4625350:13, 788974:11, 2338453:11]
 窗口[2017-11-26 15:35:00,2017-11-26 16:35:00):[2338453:20, 706266:18, 1668006:18]
 窗口[2017-11-26 17:35:00,2017-11-26 18:35:00):[3031354:8, 4443059:7, 3845720:7]
 窗口[2017-11-26 16:30:00,2017-11-26 17:30:00):[2338453:22, 1850198:22, 2364679:21]
 窗口[2017-11-26 17:50:00,2017-11-26 18:50:00):[1550635:4, 4211339:3, 1174823:3]
 窗口[2017-11-26 18:00:00,2017-11-26 19:00:00):[3715576:1, 703558:1, 3445423:1]
 窗口[2017-11-26 14:50:00,2017-11-26 15:50:00):[998153:18, 812879:18, 2364679:18]
 窗口[2017-11-26 15:10:00,2017-11-26 16:10:00):[2338453:17, 812879:16, 706266:15]
 窗口[2017-11-26 15:15:00,2017-11-26 16:15:00):[2338453:17, 706266:16, 812879:16]
 窗口[2017-11-26 15:55:00,2017-11-26 16:55:00):[2338453:24, 706266:14, 812879:14]
 窗口[2017-11-26 16:25:00,2017-11-26 17:25:00):[2338453:23, 1850198:22, 812879:17]
 窗口[2017-11-26 17:05:00,2017-11-26 18:05:00):[2338453:19, 4625350:17, 203244:16]
 窗口[2017-11-26 15:25:00,2017-11-26 16:25:00):[2338453:20, 706266:19, 812879:17]
 窗口[2017-11-26 15:45:00,2017-11-26 16:45:00):[2338453:22, 706266:17, 812879:16]
 窗口[2017-11-26 16:45:00,2017-11-26 17:45:00):[2338453:23, 2364679:21, 1850198:19]
 窗口[2017-11-26 15:00:00,2017-11-26 16:00:00):[3845720:15, 2331370:15, 812879:15]
 窗口[2017-11-26 16:05:00,2017-11-26 17:05:00):[2338453:21, 1850198:17, 812879:16]
 窗口[2017-11-26 17:25:00,2017-11-26 18:25:00):[2364679:10, 4625350:10, 2338453:10]
 窗口[2017-11-26 14:45:00,2017-11-26 15:45:00):[812879:20, 2364679:19, 3845720:16]
 窗口[2017-11-26 16:20:00,2017-11-26 17:20:00):[2338453:26, 1850198:22, 2364679:17]
 窗口[2017-11-26 16:35:00,2017-11-26 17:35:00):[1850198:23, 2364679:22, 2338453:22]
 窗口[2017-11-26 14:25:00,2017-11-26 15:25:00):[812879:20, 3845720:19, 2453685:16]
 窗口[2017-11-26 14:30:00,2017-11-26 15:30:00):[812879:19, 3845720:19, 2331370:17]
 窗口[2017-11-26 14:55:00,2017-11-26 15:55:00):[998153:17, 812879:17, 2364679:17]
 窗口[2017-11-26 14:20:00,2017-11-26 15:20:00):[812879:20, 3845720:20, 1871901:18]
 窗口[2017-11-26 16:00:00,2017-11-26 17:00:00):[2338453:26, 3845720:16, 987143:16]
 窗口[2017-11-26 17:00:00,2017-11-26 18:00:00):[2338453:21, 4625350:20, 2364679:18]
 窗口[2017-11-26 17:15:00,2017-11-26 18:15:00):[2338453:16, 4625350:15, 812879:13]
 窗口[2017-11-26 17:40:00,2017-11-26 18:40:00):[4625350:7, 3031354:6, 4714272:5]
 窗口[2017-11-26 14:35:00,2017-11-26 15:35:00):[812879:21, 3845720:20, 2364679:18]
 窗口[2017-11-26 15:05:00,2017-11-26 16:05:00):[2338453:18, 812879:15, 2331370:15]
 窗口[2017-11-26 15:20:00,2017-11-26 16:20:00):[2338453:18, 706266:17, 812879:16]
 窗口[2017-11-26 15:30:00,2017-11-26 16:30:00):[2338453:20, 812879:19, 706266:18]
 窗口[2017-11-26 17:10:00,2017-11-26 18:10:00):[2338453:18, 4625350:16, 812879:15]
 窗口[2017-11-26 14:15:00,2017-11-26 15:15:00):[3845720:20, 812879:19, 1871901:19]
 窗口[2017-11-26 14:40:00,2017-11-26 15:40:00):[812879:21, 2364679:19, 3845720:18]
 窗口[2017-11-26 16:10:00,2017-11-26 17:10:00):[2338453:22, 1850198:19, 2364679:17]
 窗口[2017-11-26 17:30:00,2017-11-26 18:30:00):[3472936:9, 3845720:8, 4625350:8]
 窗口[2017-11-26 17:45:00,2017-11-26 18:45:00):[1893705:5, 1646861:5, 3148749:5]
 窗口[2017-11-26 17:55:00,2017-11-26 18:55:00):[4443059:3, 2491788:3, 4265579:2]
 窗口[2017-11-26 15:40:00,2017-11-26 16:40:00):[2338453:20, 706266:18, 1668006:18]
 窗口[2017-11-26 15:50:00,2017-11-26 16:50:00):[2338453:22, 706266:15, 812879:15]
 窗口[2017-11-26 16:40:00,2017-11-26 17:40:00):[2364679:24, 2338453:23, 1850198:23]
 窗口[2017-11-26 16:50:00,2017-11-26 17:50:00):[2338453:25, 2364679:20, 812879:18]
 窗口[2017-11-26 16:55:00,2017-11-26 17:55:00):[2338453:22, 2364679:21, 4625350:20]
 */
public class Demo3_HotTopN
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
            .withTimestampAssigner((u, ts) -> u.getTimestamp());

        SingleOutputStreamOperator<HotTopN> ds1 = env.readTextFile("data/UserBehavior.csv").setParallelism(1)
                                                           .map(new MapFunction<String, UserBehavior>()
                                                           {
                                                               @Override
                                                               public UserBehavior map(String value) throws Exception {
                                                                   String[] words = value.split(",");
                                                                   return new UserBehavior(
                                                                       Long.valueOf(words[0]),
                                                                       Long.valueOf(words[1]),
                                                                       Integer.valueOf(words[2]),
                                                                       words[3],
                                                                       Long.valueOf(words[4]) * 1000
                                                                   );
                                                               }
                                                           })
                                                           .filter(u -> "pv".equals(u.getBehavior()))
                                                           .assignTimestampsAndWatermarks(watermarkStrategy)
                                                           //统计在每个窗口内，每个商品的点击量
                                                           .keyBy(UserBehavior::getItemId)
                                                           .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
                                                           .aggregate(new AggregateFunction<UserBehavior, Acc, HotTopN>()
                                                           {
                                                               @Override
                                                               public Acc createAccumulator() {
                                                                   return new Acc(0l, 0);
                                                               }

                                                               @Override
                                                               public Acc add(UserBehavior value, Acc acc) {
                                                                   acc.setItemId(value.getItemId());
                                                                   acc.click += 1;
                                                                   return acc;
                                                               }

                                                               @Override
                                                               public HotTopN getResult(Acc accumulator) {
                                                                   return new HotTopN(0l, 0l, accumulator.itemId, accumulator.click);
                                                               }

                                                               @Override
                                                               public Acc merge(Acc a, Acc b) {
                                                                   return null;
                                                               }
                                                               //一个窗口执行一次
                                                           }, new ProcessWindowFunction<HotTopN, HotTopN, Long, TimeWindow>()
                                                           {
                                                               @Override
                                                               public void process(Long aLong, Context context, Iterable<HotTopN> elements, Collector<HotTopN> out) throws Exception {
                                                                   TimeWindow window = context.window();
                                                                   HotTopN acc = elements.iterator().next();
                                                                   acc.setStart(window.getStart());
                                                                   acc.setEnd(window.getEnd());

                                                                   out.collect(acc);
                                                               }
                                                           });


        //再次运算，求topN
        ds1.keyBy(HotTopN::getEnd)
           //不开窗的process，一条数据执行一次
           .process(new KeyedProcessFunction<Long, HotTopN, String>()
           {

               //到点了，说明这个窗口的数据已经到齐，排序求TopN
               @Override
               public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

                   Comparator<HotTopN> comparator = (HotTopN h1, HotTopN h2) -> -h1.getClick().compareTo(h2.getClick());
                   List<HotTopN> top3 = StreamSupport.stream(list.get().spliterator(), true)
                                                        .sorted(comparator)
                                                        .limit(3)
                                                        .collect(Collectors.toList());

                   //输出结果
                   TimeWindow timeWindow = new TimeWindow(top3.get(0).getStart(), top3.get(0).getEnd());
                   String str = MyUtil.printTimeWindow(timeWindow);

                   List<String> strs = top3.stream().map(h -> h.getItemId() + ":" + h.getClick()).collect(Collectors.toList());

                   out.collect(str + ":" + strs.toString());

               }

               @Override
               public void open(Configuration parameters) throws Exception {
                   list = getRuntimeContext().getListState(new ListStateDescriptor<HotTopN>("list", HotTopN.class));
                   ifFirst = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("ifFirst", Boolean.class));
               }

               //每一条数据来，加入到一个集合
               private ListState<HotTopN> list;
               //第一条数据来，制定一个定时器，等process的watermark推进到窗口的end时，就出发
               private ValueState<Boolean> ifFirst;
               @Override
               public void processElement(HotTopN value, Context ctx, Collector<String> out) throws Exception {

                   //是第一条，注册定时器
                   if (ifFirst.value() == null){
                       //第一条已经来过
                       ifFirst.update(true);
                       //注册定时器
                       ctx.timerService().registerEventTimeTimer(value.end + 2000);
                   }

                   //每一条都要加入list
                   list.add(value);
               }
           }).print().setParallelism(1);

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @AllArgsConstructor
    @Data
    @NoArgsConstructor
    public static class Acc{
        private Long itemId;
        private Integer click;
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class HotTopN {

        private Long end;
        private Long start;
        private Long itemId;
        private Integer click;

    }
}
